package com.kasax.httpsupport;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Http执行器
 */
public class HttpExecutor {
    /**
     * 最大存活时间
     */
    private static final long KEEP_ALIVE_TIME = 10;
    private static volatile HttpExecutor instance;
    /**
     * 线程核心数
     */
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
    /**
     * 线程池
     */
    private final ThreadPoolExecutor executor;

    /**
     * 无界队列
     */
    private BlockingQueue<Future<?>> queue = new LinkedBlockingQueue<>();
    private final ThreadFactory threadFactory = new ThreadFactory() {
        AtomicInteger num = new AtomicInteger();

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, "HttpThread-" + num.incrementAndGet());
            return thread;
        }
    };


    /**
     * 拒绝策略，如果当线程池中的阻塞队列满，则添加到link队列中
     */
    private final RejectedExecutionHandler rejectHandler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                queue.put(new FutureTask<>(r, null));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    };

    /**
     * 消费者：执行任务
     */
    private Runnable runnable = new Runnable() {
        @Override
        public void run() {

            while (true) {
                FutureTask<?> task = null;
                try {
                    task = (FutureTask<?>) queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (task != null) {
                    executor.execute(task);
                }
            }
        }
    };

    private HttpExecutor() {
        executor = new ThreadPoolExecutor(CORE_POOL_SIZE, CORE_POOL_SIZE + 1, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue<>(4),
                threadFactory, rejectHandler);
        executor.execute(runnable);
    }

    public static HttpExecutor getInstance() {
        if (instance == null) {
            synchronized (HttpExecutor.class) {
                if (instance == null) {
                    instance = new HttpExecutor();
                }
            }
        }
        return instance;
    }

    /**
     * 同步执行任务
     */
    public synchronized Response submit(Callable<Response> task) throws ExecutionException, InterruptedException {
        if (task == null) {
            throw new NullPointerException("task == null");
        }
        Future<Response> future = executor.submit(task);
        return future.get();
    }

    /**
     * 异步执行任务
     */
    public void enqueue(FutureTask<?> task) {
        if (task == null) {
            throw new NullPointerException("task == null");
        }
        try {
            queue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
